package com.navinfo.tripanalysis.offline.service.impl.huashen;

import com.navinfo.tripanalysis.common.pojo.Point0F3D;
import com.navinfo.tripanalysis.common.pojo.RunningType;
import com.navinfo.tripanalysis.common.pojo.RunningTypeEnum;
import com.navinfo.tripanalysis.common.util.CommonUtils;
import com.navinfo.tripanalysis.offline.pojo.LoadDataParam;
import com.navinfo.tripanalysis.offline.service.LoadPoint0F3DService;
import com.navinfo.tripanalysis.offline.service.impl.AbstractLoadPoint0F3DService;
import com.navinfo.tripanalysis.offline.util.PointUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

/**
 * 通过hive加载0F3D报警事件明细数据
 * @author hmc
 */
@RunningType(value = RunningTypeEnum.HS,interfaceClass = LoadPoint0F3DService.class)
public class LoadPoint0F3DServiceHiveImpl extends AbstractLoadPoint0F3DService {

    public LoadPoint0F3DServiceHiveImpl() {
        COL_NAMES.addAll(Arrays.asList(TID_NAME, "gpstime", "endgpstime", "incidentid"
                ));
    }
    @Override
    protected String getWhereSql(LoadDataParam param) {
        List<Long> tidList = param.getTidList();

        //分批处理
        String batchWhere = PointUtils.getBatchWhereSqlWihtoutHashtid(param);
        StringBuilder sb = new StringBuilder(StringUtils.isEmpty(batchWhere) ? "" : batchWhere+" and ")
                .append(" gpstime!=0 and endgpstime !=0 and gpstime is not null and "+ TID_NAME +"!=0 and "+ TID_NAME +" is not null ")
                .append(CollectionUtils.isNotEmpty(tidList) ? String.format(" and %s in(%s) ", TID_NAME, CommonUtils.mkString(tidList, ",")) : "");

        return sb.toString();
    }

    @Override
    protected Point0F3D getPoint(Row row) {

        Point0F3D point = new Point0F3D();
        point.setTerminalid(Long.parseLong(Optional.ofNullable(row.getAs("terminalid")).orElse("0").toString()));
        point.setGpstime(Long.parseLong(Optional.ofNullable(row.getAs("gpstime")).orElse("0").toString()));
        point.setEndgpstime(Long.parseLong(Optional.ofNullable(row.getAs("endgpstime")).orElse("0").toString()));
        point.setIncidentid(Integer.parseInt(Optional.ofNullable(row.getAs("incidentid")).orElse("0").toString()));
        return point;
    }

    @Override
    public JavaPairRDD<Long, List<Point0F3D>> sortAndGroup(SparkSession spark, JavaSparkContext jsc, LoadDataParam param) {
        JavaRDD<Point0F3D> points0f3d = load(spark,jsc,param);
        if(points0f3d == null){
            logger.error("OOO:0f3d data is null");
            points0f3d = jsc.parallelize(new ArrayList<>());
        }
        JavaPairRDD<Long,Point0F3D> tidPoints0F3D =
                points0f3d.mapPartitionsToPair((PairFlatMapFunction<Iterator<Point0F3D>, Long, Point0F3D>) iterator ->
                        new Iterator<Tuple2<Long, Point0F3D>>() {
                            @Override
                            public boolean hasNext() {
                                return iterator.hasNext();
                            }

                            @Override
                            public Tuple2<Long, Point0F3D> next() {
                                Point0F3D point = iterator.next();
                                return new Tuple2<>(point.getTerminalid(),point);
                            }
                        });
        JavaPairRDD<Long,List<Point0F3D>> rtn =
                tidPoints0F3D.combineByKey(
                        (Function<Point0F3D, List<Point0F3D>>) point -> Stream.of(point).collect(Collectors.toList()),
                        (Function2<List<Point0F3D>, Point0F3D, List<Point0F3D>>) (points, p) -> {
                            points.add(p);
                            return points;
                        },
                        (Function2<List<Point0F3D>, List<Point0F3D>, List<Point0F3D>>)(p1,p2) -> {
                            p1.addAll(p2);
                            return p1;
                        })
                        .mapPartitionsToPair((PairFlatMapFunction<Iterator<Tuple2<Long, List<Point0F3D>>>, Long, List<Point0F3D>>) tuple ->
                                new Iterator<Tuple2<Long, List<Point0F3D>>>() {
                                    @Override
                                    public boolean hasNext() {
                                        return tuple.hasNext();
                                    }

                                    @Override
                                    public Tuple2<Long, List<Point0F3D>> next() {
                                        Tuple2<Long,List<Point0F3D>> tuple2 = tuple.next();
                                        long tid = tuple2._1;
                                        List<Point0F3D> points = tuple2._2;
                                        List<Point0F3D> sortedPoints = null;
                                        if(!CommonUtils.isCollectionEmpty(points)){
                                            sortedPoints = StreamSupport.stream(points.spliterator(), false)
                                                    .sorted(Point0F3D.COMPARATOR).collect(Collectors.toList());
                                        }
                                        return new Tuple2<>(tid,sortedPoints);
                                    }
                                });
        return rtn;
    }
}
